package bolts;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.PrintWriter;
import java.text.SimpleDateFormat;
import java.util.Map;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;

public abstract class AbstractOutputToFileBolt extends BaseBasicBolt {

    protected PrintWriter writer;

    protected SimpleDateFormat dateFormat;

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        try {
            this.dateFormat = new SimpleDateFormat((String) stormConf.get("dateFormat"));
            this.writer = new PrintWriter(new File(getOutputFileName(stormConf)));
        } catch (FileNotFoundException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void cleanup() {
        if (writer != null) {
            writer.flush();
            writer.close();
        }
    }

    protected abstract String getOutputFileName(Map stormConf);
}
